package cn.wzy.rocketmq.listener;

import cn.hutool.core.collection.CollectionUtil;
import cn.wzy.constants.GiftConstants;
import cn.wzy.domain.Spu;
import cn.wzy.dto.GoodsSpuOnsaleDetailDocDto;
import cn.wzy.service.ISpuService;
import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.baomidou.mybatisplus.mapper.Wrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.util.List;
import java.util.stream.Collectors;

@Component
@RocketMQTransactionListener(txProducerGroup =
        GiftConstants.ONSALE_TX_PRODUCER_GROUP)
@Slf4j
public class OnsaleLocalTransactionListener implements RocketMQLocalTransactionListener {

    @Autowired
    private ISpuService spuService;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        try {
            //获取消息内容json字符串
            byte[] payload = (byte[]) message.getPayload();
            String jsonStr = new String(payload, Charset.forName("utf-8"));
            System.out.println(payload);
            //转换为dto
            List<GoodsSpuOnsaleDetailDocDto> docDtos = JSONArray.parseArray(jsonStr,
                    GoodsSpuOnsaleDetailDocDto.class);

            //进而获取id
            List<Long> ids = docDtos.stream()
                    .map(dto -> dto.getId()).collect(Collectors.toList());
            spuService.onsale(ids);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("上架失败!");
            return RocketMQLocalTransactionState.ROLLBACK;
        }

    }


    //会查状态
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //获取消息内容json字符串
        String jsonStr = (String) message.getPayload();

        //转换为dto
        List<GoodsSpuOnsaleDetailDocDto> docDtos = JSONArray.parseArray(jsonStr,
                GoodsSpuOnsaleDetailDocDto.class);

        //只要有一个是上架状态就认为执行成功
        List<Long> ids = docDtos.stream().map(dto -> dto.getId()).collect(Collectors.toList());
        Wrapper<Spu> wrapper = new EntityWrapper<Spu>()
                .in("id", ids)
                .and()
                .eq("state", 2);

        List<Spu> goodsSpus = spuService.selectList(wrapper);
        if (CollectionUtil.isNotEmpty(goodsSpus))
            return RocketMQLocalTransactionState.COMMIT;

        return RocketMQLocalTransactionState.ROLLBACK;
    }

}